/****************************************************************************
 *
 * Copyright (C) 2003-2008 Los Alamos National Security, LLC
 *                         Packet Analytics Corporation
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License Version 2 as
 * published by the Free Software Foundation.  You may not use, modify or
 * distribute this program under any other version of the GNU General
 * Public License.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 *
 ****************************************************************************/
package nfse.gui.server;

import java.io.DataInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.PrintWriter;
import java.net.ServerSocket;

import javax.net.*;

import org.postgresql.PGConnection;

import java.net.*;
import java.sql.*;

import nfse.NetFSE;
import nfse.NetFSERecord;
import nfse.NetFSEUtil;

/**
 * This class is derived from QueryThread. It is essentially the same but
 * instead of inserting results one at a time it creates a file that is then
 * processed using PostgreSQL's COPY command.
 * 
 * @author bduphoff
 * 
 */
public class QueryThreadCopy extends Thread {

    String queryID = "";

    String[] queries = null;

    int[] types = null;

    Connection connection = null;

    long total = 1000000;

    long limit = 1000000;

    PrintWriter fieldOut = null;

    PrintWriter recordOut = null;

    File fieldFile = null;

    File recordFile = null;

    public QueryThreadCopy(String queryID, String[] queries, int[] types, Connection connection, long total, long limit) {
        this.queryID = queryID;
        this.queries = queries;
        this.types = types;
        this.connection = connection;
        this.total = total;
        this.limit = limit;
    }

    public void run() {
        long startTS = System.currentTimeMillis();
        Socket socket = null;
        Socket socket2 = null;
        DataInputStream resultIn = null;
        Statement statement = null;
        try {
            statement = connection.createStatement();

            String insertSQL = "";
            ResultSet rs = null;
            for (int x = 0; x < queries.length; x++) {
                // SocketFactory sf = SSLSocketFactory.getDefault();
                SocketFactory sf = SocketFactory.getDefault();
                socket = sf.createSocket(NetFSE.QUERY_SERVER_HOST, NetFSE.QUERY_SERVER_PORT);
                PrintWriter socketOut = new PrintWriter(socket.getOutputStream());
                DataInputStream socketIn = new DataInputStream(socket.getInputStream());

                ServerSocketFactory ssf = ServerSocketFactory.getDefault();
                ServerSocket ss = ssf.createServerSocket();
                try {

                    if ((queries[x].indexOf(" user=\"") > 0) && (queries[x].indexOf(" password=\"") > 0)) {
                        // no op
                    }

                    int endIdx = queries[x].lastIndexOf("</Query>");
                    if (endIdx >= 0) {
                        queries[x] = queries[x].substring(0, endIdx);
                    } else {
                        endIdx = queries[x].lastIndexOf("/>");
                        queries[x] = queries[x].substring(0, endIdx) + ">";
                    }

                    ss.bind(null);

                    queries[x] += "<SocketResult host=\"" + NetFSE.LOCAL_HOST + "\" port=\"" + ss.getLocalPort()
                            + "\"/>";

                    queries[x] += "</Query>";

                    // long start = System.currentTimeMillis();
                    socketOut.println(queries[x]);
                    socketOut.println("<--DiSARM: end query-->");
                    socketOut.flush();
                } catch (Exception e) {
                    e.printStackTrace();
                }
                // out.println(queries[x] + "<br>");

                int reply = socketIn.readInt();
                if (reply >= 0)
                    socket2 = ss.accept();

                if (reply < 0) {
                    // statement.close();
                    return;
                } else {

                    resultIn = new DataInputStream(socket2.getInputStream());
                    // long start = System.currentTimeMillis();
                    int count = 0;

                    int type = -1;
                    int sensor = -1;
                    int len = -1;
                    byte[] bytes = null;
                    int bytesRead;
                    String line = null;
                    try {
                        type = resultIn.readInt();
                        if (type == -100) {
                            line = null;
                        } else {
                            sensor = resultIn.readInt();
                            len = resultIn.readInt();
                            bytes = new byte[len];
                            bytesRead = 0;
                            while (bytesRead < len)
                                bytesRead += resultIn.read(bytes, bytesRead, len - bytesRead);
                            // for (int j = 0; j < len; j++)
                            // bytes[j] = (byte) resultIn.read();

                            line = new String(bytes);
                        }
                    } catch (Exception E) {
                        E.printStackTrace();
                        line = null;
                    }
                    insertSQL = "";
                    GUIRecordFactory factory = null;
                    try {
                        factory = (GUIRecordFactory) NetFSE.getRecordFactory(type);
                    } catch (Exception e) {
                        line = null;
                        e.printStackTrace();
                    }
                    NetFSERecord temp = null;
                    int typeCount = 0;

                    while (line != null) {

                        if ((total != -1) && (count == total)) {
                            line = null;
                            break;
                        }

                        if ((limit == -1) || (typeCount < limit)) {
                            try {
                                // System.out.println(line);
                                temp = (NetFSERecord) factory.generate(line, sensor);
                                if (temp != null) {
                                    String fieldSQL = factory.getInsertFieldTableCopySQL(temp, count);
                                    String recordSQL = count + "\t" + type + "\t" + sensor + "\t"
                                            + NetFSEUtil.convertToHex(line) + "";
                                    insert(fieldSQL, recordSQL);
                                    typeCount++;
                                    count++;
                                }
                            } catch (Exception e) {
                                System.out.println("FAILED: " + insertSQL);
                                e.printStackTrace();
                            }
                        } else {
                            line = null;
                        }

                        /**
                         * After 1000 records, process them
                         */
                        if (count % 1000 == 0) {
                            copy(queryID + "_" + type, queryID);
                        }

                        if (line != null) {
                            try {
                                type = resultIn.readInt();
                                if (type == -100) {
                                    line = null;
                                } else {
                                    sensor = resultIn.readInt();
                                    len = resultIn.readInt();
                                    bytes = new byte[len];
                                    bytesRead = 0;
                                    while (bytesRead < len)
                                        bytesRead += resultIn.read(bytes, bytesRead, len - bytesRead);

                                    // for (int j = 0; j < len; j++)
                                    // bytes[j] = (byte) resultIn.read();
                                    line = new String(bytes);
                                }
                            } catch (Exception E) {
                                E.printStackTrace();
                                line = null;
                            }
                        }

                    } // End main while loop

                    /**
                     * Process any records that were written out but not
                     * processed within the loop.
                     */
                    copy(queryID + "_" + type, queryID);

                }

                rs = statement.executeQuery("select count(*) as a from " + queryID + " where Type_ID=" + types[x]);
                rs.next();
                int cnt = rs.getInt("a");
                rs.close();
                statement.execute("update Query_Types set Count=" + cnt + " where Query_ID='" + queryID
                        + "' and Type_ID=" + types[x]);

                if (resultIn != null) {
                    resultIn.close();
                }
                socketOut.close();
                socketIn.close();
                if (socket2 != null)
                    socket2.close();
                socket.close();
                ss.close();
            }

            rs = statement.executeQuery("select count(*) as a from " + queryID);
            rs.next();
            int cnt = rs.getInt("a");
            rs.close();
            statement
                    .execute("update Queries set Count=" + cnt + ", Status=" + 1 + " where Query_ID='" + queryID + "'");

            // c.close();
        } catch (Exception E) {
            E.printStackTrace();
        } finally {
            try {
                statement.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
            long endTS = System.currentTimeMillis();
            long timeInMS = endTS - startTS;
            System.out.println("Query completed in " + timeInMS/1000 + " seconds.");
        }
    }

    /**
     * Writes insert information for fields and records to disk. If files are
     * null, creates new temp files, opens output streams.
     * 
     * @param fieldSQL
     *            The field values for a single record in tab-delimited format
     *            for PostgreSQL's COPY command
     * @param recordSQL
     *            The record values for a single record in tab-delimited format
     *            for PostgreSQL's COPY command
     */
    private void insert(String fieldSQL, String recordSQL) throws IOException {
        if (recordFile == null) {
            recordFile = File.createTempFile("record", ".sql");
            fieldFile = File.createTempFile("field", ".sql");
            recordOut = new PrintWriter(recordFile);
            fieldOut = new PrintWriter(fieldFile);
        }
        recordOut.println(recordSQL);
        fieldOut.println(fieldSQL);
    }

    /**
     * Called after main loop in run() or by insert after a fixed number of
     * results have been written to disk. Calls COPY on the field and record
     * files. Delets old temp files, closes output streams, sets all to null;
     */
    private void copy(String fieldTable, String recordTable) throws IOException, SQLException {
        /**
         * If recordOut is null this means that there is nothing to COPY,
         * return.
         */
        if (recordOut == null)
            return;

        /**
         * Flush and close the output streams in preparation of COPY.
         */
        recordOut.flush();
        recordOut.close();
        fieldOut.flush();
        fieldOut.close();

        FileInputStream recordIn = new FileInputStream(recordFile);
        ((PGConnection) connection).getCopyAPI().copyIntoDB("COPY " + recordTable + " FROM STDIN", recordIn);
        recordIn.close();

        
        /**
         * The code below is needed due to some bugs in PostgreSQL's handling of
         * UTF-8 encoding. This command will strip the bad unicode out of the
         * file so that the COPY command will run.
         * 
         * Testing revealed that this code did not in fact work reliably. The fix was to allow
         * fields to be hex-encoded. These fields can't be searched (except in hex) but that is
         * deemed acceptable.
         */
        /*
        File tempFieldFile = File.createTempFile("tempFieldFile", ".sql");
        File tempFieldFileSH = File.createTempFile("tempFieldFile", ".sh");
        String utf8fix = "iconv -c -f UTF-8 -t UTF-8 " + fieldFile.getAbsolutePath() + " > "
                + tempFieldFile.getAbsolutePath();
        PrintWriter out = new PrintWriter(tempFieldFileSH);
        out.println(utf8fix);
        out.flush();
        out.close();
        NetFSEUtil.execAndWait("sh " + tempFieldFileSH.getAbsolutePath());
        fieldFile.delete();
        tempFieldFile.renameTo(fieldFile);
        tempFieldFile.delete();
        tempFieldFileSH.delete();
        */
        
        FileInputStream fieldIn = new FileInputStream(fieldFile);
        ((PGConnection) connection).getCopyAPI().copyIntoDB("COPY " + fieldTable + " FROM STDIN", fieldIn);
        fieldIn.close();

        
        /**
         * Delete the temp files that were just processed, null out the
         * references so that insert() will create new files and output streams.
         */
        recordOut = null;
        fieldOut = null;
        recordFile.delete();
        recordFile = null;
        fieldFile.delete();
        fieldFile = null;
    }
}
